package system;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.rmi.Naming;
import java.rmi.RMISecurityManager;
import java.rmi.RemoteException;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import api.TaskBase;
/**
 * The class for the Worker process. It implements the Worker interface
 * which is used by the Master process to communicate with the Worker.
 * 
 * @author Mayuri Karra, Bing Wei
 *
 */
public class WorkerImpl extends UnicastRemoteObject implements Worker{
	
	private static final long serialVersionUID = 1L;
	int workerId;
	String mapOutputFileName;
	String reduceInputFileName;
	String reduceOutputFileName;
	String lookupInputFileName;
	static boolean doSleep = false;
	
	/**
	 * getter function for the worker id
	 */
	public int getWorkerId() throws RemoteException{
		return workerId;
	}
	
	/**
	 * A method for setting the file names used by this class
	 * 
	 * @param taskId
	 * @param subTaskId
	 */

	public void iniFileNames(long taskId, long subTaskId) {		
	  mapOutputFileName = "mapper_"+  String.valueOf(taskId)+"_" + String.valueOf(subTaskId)+"_output.txt";
	  reduceInputFileName = "reducer_"+  String.valueOf(taskId)+"_"+ String.valueOf(subTaskId)+"_input.txt";		
	  reduceOutputFileName = "reducer_"+  String.valueOf(taskId)+"_"+ String.valueOf(subTaskId)+"_output.txt";		
	}
	
	protected WorkerImpl() throws RemoteException {
		super();
	}

	/**
	 * Helper function for generating output file names for shuffle
	 * @param i
	 * @param taskId
	 * @param subTaskId
	 * @return a file name which is an encoding of the given input values
	 */
	String getShuffleOutputFileName(int i, long taskId, long subTaskId){
		return ("reducer_" + taskId + "_" + subTaskId + "_" +  i + "_input.txt");	
	}

	/**
	 * Helper function for generating input file names for reduce phase
	 * @param i
	 * @param taskId
	 * @param subTaskId
	 * @returna file name which is an encoding of the given input values
	 */
	String getReducerInputFileName(int i, long taskId, long subTaskId){
		return ("reducer_" + taskId + "_" + i + "_" +  subTaskId + "_input.txt");	
	}

	/**
	 * The method to execute a given task passed by the Master.
	 * The task can be map/reduce
	 * 
	 * @param masterTask
	 * @return the time taken by this task to execute
	 * @throws RemoteException
	 */

	@Override
	public long execute(MasterTaskBase masterTask) throws RemoteException{
		
		if (doSleep)
			try {
				System.out.println("about to sleep for 10 seconds");
				Thread.sleep(10000);
				System.out.println("done sleeping");
			} catch (InterruptedException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}
		
		long startTime = System.currentTimeMillis();
		iniFileNames (masterTask.getClientTask().getTaskId(), masterTask.getSubTaskId());
		if(masterTask.getType() == MasterTaskBase.MAPPER){
			TaskBase clientTask = masterTask.getClientTask();
			BufferedReader input = null;
			BufferedWriter out = null;
			int numWorkers = masterTask.getNumSubTasks();
		    BufferedWriter[] shuffleOut = new BufferedWriter[numWorkers];

			try {
				input =  new BufferedReader(new FileReader(clientTask.getInputFileName()));
				out = new BufferedWriter(new FileWriter(mapOutputFileName));
				for (int i = 0; i < numWorkers; i++){
					shuffleOut[i] = new BufferedWriter(new FileWriter(getShuffleOutputFileName(i, clientTask.getTaskId(), masterTask.getSubTaskId())));
				}		

				
		        String line = null; //not declared within while loop
		        int lineNum = -1;
		        
		        //TODO change the read
	        	while (( line = input.readLine()) != null){
	        		lineNum ++;
		            if(lineNum >=masterTask.getIndexLow() && lineNum <= masterTask.getIndexHigh()){
		            	// TODO the API have no idea what to give as key and value to mapper, this is application specific
		            	List<MapperResult> mapRst = masterTask.getClientTask().map(line, lineNum);
		            	for(MapperResult rst: mapRst){
		            		out.write(rst.getKey()+"\t"+rst.getValue()+"\n");
		              		shuffleOut[Math.abs((rst.getKey()).hashCode())%numWorkers].write(rst.getKey()+"\t"+rst.getValue()+"\n");		      		            	
		            	}
		            }
	        	}
	        	input.close();
	        	out.close();
				for (int i = 0; i < numWorkers; i++){
					shuffleOut[i].close();
				}		
	        }
		    catch (IOException ex){
		      ex.printStackTrace();
		    }
		    finally {
		        try {
					input.close();
					out.close();
					for (int i = 0; i < numWorkers; i++){
						shuffleOut[i].close();
					}		
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		        
		    }			
			
		}
		else if(masterTask.getType() == MasterTaskBase.REDUCER){
			TaskBase clientTask = masterTask.getClientTask();
			BufferedWriter out = null;
			int numWorkers = masterTask.getNumSubTasks();
		    BufferedReader[] reducerIn = new BufferedReader[numWorkers];
			System.out.println("Reducer task  numWorkers " + numWorkers);
	
			try {
				out = new BufferedWriter(new FileWriter(reduceOutputFileName));			
	            Map<Object,List<Object>> hashMap = new HashMap <Object, List<Object>>();
				for (int i = 0; i < numWorkers; i++){
					reducerIn[i] = new BufferedReader(new FileReader(getReducerInputFileName(i, clientTask.getTaskId(), masterTask.getSubTaskId())));
					System.out.println("Reducer input file " + getReducerInputFileName(i, clientTask.getTaskId(), masterTask.getSubTaskId()));
				}		    
	            
		        String line = null; //not declared within while loop
		        
		        //TODO change the read
				for (int i = 0; i < numWorkers; i++){
					System.out.println("Reducer input file " + getReducerInputFileName(i, clientTask.getTaskId(), masterTask.getSubTaskId()));
					
		        	while (( line = reducerIn[i].readLine()) != null){
		        		try{
			        	String[] strs = line.split("\t");
			        	
		                if (hashMap.containsKey(strs[0])){
		                	List<Object> valueList = hashMap.get(strs[0]);
		                	valueList.add(strs[1]);
		                	hashMap.put(strs[0],valueList);
		                	
		                }else{
		                	List<Object> valueList = new ArrayList<Object>();
		                	valueList.add(strs[1]);
		                  	hashMap.put(strs[0], valueList);                		                	
		                }	     
		        		}catch(Exception e){
		        			;
		        			//System.out.println ("Error in reduction " +  getReducerInputFileName(i, clientTask.getTaskId(), masterTask.getSubTaskId()));
		        			//System.out.println ("line is " +  line );		        			
		        		}
		        	}		        	
				}
	        	
	        	for(Map.Entry<Object, List<Object>> keyValues: hashMap.entrySet()){
	        		ReducerResult rst = clientTask.reduce(keyValues.getKey(), keyValues.getValue());
	        		out.write(rst.getKey()+"\t"+rst.getValue()+"\n");
	        	}

	        	out.close();
	        }
		    catch (IOException ex){
		      ex.printStackTrace();
		    }
		    finally {
		        try {
					out.close();
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
		        
		    }
			
		}
		
		long endTime = System.currentTimeMillis();
		return endTime - startTime;
		
	}
	
	/**
	 * Method to lookup a given key in the output of reduce phase.
	 * Invoked by Master when a client asks the Master for the key
	 * @param masterTask
	 * @return the result as a String
	 * @throws RemoteException
	 */
	@Override
	public String lookup(MasterTaskBase masterTask) throws RemoteException{
		iniFileNames (masterTask.getTaskId(), masterTask.getSubTaskId());
		String lookup = masterTask.getLookupKey();
		BufferedReader input = null;
		try {
			input =  new BufferedReader(new FileReader(reduceOutputFileName));
			System.out.println("Looking up for " + lookup + " in " + reduceOutputFileName);
	        String line = null; //not declared within while loop
	        
	        //TODO change the read
        	while (( line = input.readLine()) != null){
	        	String[] strs = line.split("\t");
	        	
                if(strs[0].equals(lookup)){
                    System.out.println("done with looking up for " + lookup + " result " + strs[1]);
                	return strs[1];
                }
        	}
            System.out.println("done with looking up for " + lookup + " no result ");
        	input.close();
        	
        }
	    catch (IOException ex){
	      ex.printStackTrace();
	    }
	    return null;

	}

	public static void main(String args[]) throws Exception {

		String masterDomainName = args[0];
		int sleep = Integer.parseInt(args[1]);
		
		doSleep = (sleep > 0)? true:false;
		
		if (System.getSecurityManager() == null) {
			System.setSecurityManager(new RMISecurityManager());
		}

		Worker2Master masterServer = (Worker2Master) Naming.lookup("//"
				+ masterDomainName + "/" + Worker2Master.SERVICE_NAME);
        InetAddress addr = InetAddress.getLocalHost();
        
		Worker worker = new WorkerImpl();
		String processStr = ManagementFactory.getRuntimeMXBean().getName();
		String processId = processStr.substring(0, processStr.indexOf('@'));
		
		masterServer.register(worker, addr.getHostName(), Integer.valueOf(processId));
		System.out.println("Worker registered !!");
	}
	
	@Override
	public void getPing() throws RemoteException{
		;
	}
}
